package com.zn.socketReceiver;

import com.alibaba.fastjson.JSONObject;
import com.zn.entity.OptionMarketDataField;
import com.zn.service.MdInfoService;
import com.zn.util.*;
import com.zn.vo.EtfMdInfo;
import com.zn.vo.OptionSymbol;
import com.zn.vo.SocketJsonData;
import com.zn.web.rpc.MdRemoteCall;
import com.zn.websocket.group.ChannelGroups;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Created by yinjiawei on 2018/7/12.
 */
public class ReaderTask extends Thread{
    private static final Logger LOGGER = LoggerFactory.getLogger(ReaderTask.class);
    private SocketStatusListener socketStatusListener;
    private static final Integer FIDLE_SIZE=30;
    private RedisTemplate<String,String> redisTemplate;
    private MongoTemplate mongoTemplate;
    private MdInfoService mdInfoService;
    private MdRemoteCall mdRemoteCall;
    private ExecutorService pool = Executors.newCachedThreadPool();
    private static final List<String> EX_DATE_NUM = Arrays.asList("A", "B", "C","D");
    private Socket socket;

    private boolean listening;

    public ReaderTask(Socket socket,RedisTemplate<String,String> redisTemplate,
                      MongoTemplate mongoTemplate,MdInfoService mdInfoService,MdRemoteCall mdRemoteCall) throws IOException{
        this.socket = socket;
        this.redisTemplate=redisTemplate;
        this.mongoTemplate=mongoTemplate;
        this.mdInfoService=mdInfoService;
        this.mdRemoteCall=mdRemoteCall;
    }

    /**
     * finish:(这里用一句话描述这个方法的作用). <br/>
     * TODO(这里描述这个方法适用条件 – 可选).<br/>
     * @throws IOException
     *
     */
    public void finish() throws IOException
    {
        listening = false;
        interrupt();
        if(socket!=null)
        {
            if(socket.isInputShutdown())
            {
                socket.shutdownInput();
            }
        }
    }

    /* (non-Javadoc)
     * @see java.lang.Runnable#run()
     */
    @Override
    public synchronized  void run()
    {
        while (listening)
        {
            try {
                this.process();
            } catch (Exception e) {
                listening = false;
                if(socketStatusListener!=null)
                {
                    int status = parseSocketStatus(e);
                    socketStatusListener.onSocketStatusChanged(socket, status, e);
                }
                e.printStackTrace();
                return;//终止线程继续运行,这里也可以使用continue
            }

        }
    }

    public void process() throws Exception {
        // Prepare date format
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd HH:mm:ss.SSS");
        dateFormat.setTimeZone(Constant.DEFAULT_TIMEZONE);
        // Prepare read buffer
        ByteBuffer bufReceive = ByteBuffer.allocate(1024 * 200);
        byte bufRead[] = new byte[1024 * 100];
        InputStream is = socket.getInputStream();
        try {
            int count = 0;
            while((count = is.read(bufRead)) > 0) {
                if(count > bufReceive.remaining()) {
                    throw new Exception("Proxy data receiver buffer overflow");
                }
                bufReceive.put(bufRead, 0, count);
                this.parseData(bufReceive);
            }
        }
        finally {
            try {is.close();}catch(Exception e) {e.printStackTrace();}
            try {socket.close();}catch(Exception e) {e.printStackTrace();}
        }
    }

    /**
     * 解析行情数据
     * @param buf 数据缓冲
     */
    protected void parseData(ByteBuffer buf) {
        buf.flip();
        byte data[] = new byte[buf.remaining()];
        buf.get(data);
        int startIndex = 0;
        for(int i = 0; i < data.length; i++) {
            if(data[i] == '\n') {
                if(i > startIndex) {
                    String item = null;
                    try {
                        item = new String(data, startIndex, i - startIndex, "GB2312");
                        if(item.startsWith("heartbeat")) {//接受心跳数据

                        }else if(item.startsWith("optionSymbol")) {
                            //接受主力合约列表的数据
							LOGGER.warn("optionSymbol"+item);
                            this.parseOptionSymbol(item);
                        }else if(item.startsWith("tick")){
                            //接受深度行情数据
							LOGGER.info("tick" + item);
                            this.parseDataItem(item);
                        }
                    }catch(Exception e) {
                        e.printStackTrace();
                        LOGGER.error(this.getClass() + "Proxy data receiver parse error: " + e.getMessage());
                        LOGGER.error(this.getClass() + "" + item);
                    }
                }
                startIndex = i + 1;
            }
        }
        buf.clear();
        if(startIndex < data.length) {
            buf.put(data, startIndex, data.length - startIndex);
        }
    }

    /**
     * 每天推一次的合约信息
     * @param itme
     */
    public void parseOptionSymbol(String itme){
        //optionSymbol|合约|合约名称|交割年份|交割月|201811|行权价|购(1)/沽(2)|基础商品代码|合约到期日
        //optionSymbol|10001423|50ETF购3月2700|2019|03|201903|2.7000|1|510050|2019-03-27|
        String [] symbolArray=itme.split("\\|");
        String instrumentCode=symbolArray[1],instrumentName=symbolArray[2],deliveryYear=symbolArray[3],deliveryMonth=symbolArray[4],
                instrumentMonth=symbolArray[5],execsPrice=symbolArray[6],instrumentType=symbolArray[7],targetCode=symbolArray[8],expireDay=symbolArray[9];
        String nowDate=DateUtils.convertDateToString(new Date());
//        String nowDate=DateUtils.convertDateToString(DateUtils.minusDay(new Date(), 1));
        try {
            /**开头如果不为50etf 如：XD50ETF购12月2250**/
            String newInstrumentName="",newExecsPrice="";
            newInstrumentName=instrumentName.replace("XD","").replace("XR","").replace(" ","");
            String exDataValue=newInstrumentName.substring(newInstrumentName.length()-1);
            /**截取合约名称中的除权次数 A|B|C|D**/
            if(EX_DATE_NUM.contains(exDataValue)){
                newExecsPrice=execsPrice+exDataValue;
                execsPrice=newExecsPrice;
            }else{
                newExecsPrice=execsPrice;
            }
            String redisKey=newInstrumentName+"-"+newExecsPrice;
            /**校验当前的 50ETF购12月2500 是否在去重合约中存在**/
            String beFrontInstrumentCode=(String)redisTemplate.opsForHash().get(Constant.DISTINCT_INSTRUMENT+nowDate,redisKey);
            /**如果不存在**/
            if(StringUtils.isBlank(beFrontInstrumentCode)){
                /**将当前的 合约名称+行权价的key 存入去重集合中**/
                redisTemplate.opsForHash().put(Constant.DISTINCT_INSTRUMENT+nowDate,redisKey,instrumentCode);
            }else{
                /**如果存在，则查看已经入redis的code在昨天是否出现**/
                String yesterday=DateUtils.convertDateToString(DateUtils.minusDay(new Date(),1));
                String optionSymbolKey=this.checkOptionKey(yesterday,10);
                Object beFrontOptionSymbolObj=redisTemplate.opsForHash().get(optionSymbolKey,beFrontInstrumentCode);
                /**如果已经入去重集合的这个合约在上一个交易日没有出现**/
                if(beFrontOptionSymbolObj==null){
                    /**则将已入当日合约集合的合约code移除，**/
                    LOGGER.info("optionSymbolDelete----->" + beFrontInstrumentCode);
                    redisTemplate.opsForHash().delete(Constant.OPTION_SYMBOL_INFO + nowDate, beFrontInstrumentCode);
                    /**去重集合中覆盖为昨日存在的合约**/
                    redisTemplate.opsForHash().put(Constant.DISTINCT_INSTRUMENT + nowDate, redisKey, instrumentCode);
                }else{
                    /**如果昨天有数据 则当前key是新的合约，不存入option-symbol中**/
                    LOGGER.info("optionSymbolReturn----->" + itme);
                    return ;
                }
            }
            /**存标的信息**/
            boolean targetFlag=redisTemplate.opsForHash().hasKey(Constant.TARGET_CODE_LIST,targetCode);
            if(!targetFlag){
                /**如果标的信息不存在 则去查询**/
                String etfMdInfoApiStr=mdRemoteCall.getEtfMdInfo(targetCode);
                String [] etfMdInfos=subStringSinaData(etfMdInfoApiStr);
                if(etfMdInfos!=null){
                    EtfMdInfo etfMdInfo=new EtfMdInfo(etfMdInfos[0],etfMdInfos[1],etfMdInfos[2],etfMdInfos[3],etfMdInfos[4],etfMdInfos[5]);
                    Object etfJson = JSONObject.toJSON(etfMdInfo);
                    redisTemplate.opsForHash().put(Constant.TARGET_CODE_LIST, targetCode, etfJson.toString());
                }
            }
            /**存标的月份 且保存7天**/
            redisTemplate.opsForSet().add(Constant.INSTRUMENT_MONTH_SET+targetCode+":"+nowDate,instrumentMonth);
            redisTemplate.expire(Constant.INSTRUMENT_MONTH_SET + targetCode+":"+nowDate, 10, TimeUnit.DAYS);
            /**存行权价 且保存7天**/
            redisTemplate.opsForSet().add(Constant.EXECS_PRICE_SET+targetCode+":"+instrumentMonth+":"+nowDate,execsPrice);
            redisTemplate.expire(Constant.EXECS_PRICE_SET+targetCode+":"+instrumentMonth+":"+nowDate, 10, TimeUnit.DAYS);
            /**存Symbol信息**/
            OptionSymbol optionSymbol=new OptionSymbol(instrumentCode,instrumentName.replace(" ",""),deliveryYear,deliveryMonth,
                    instrumentMonth,execsPrice,instrumentType,targetCode,expireDay);
            Object optionSymbolJson = JSONObject.toJSON(optionSymbol);
            /**key规则：OPTION-SYMBOL-INFO:2018-11-06**/
            redisTemplate.opsForHash().put(Constant.OPTION_SYMBOL_INFO + nowDate, instrumentCode, optionSymbolJson.toString());
            redisTemplate.expire(Constant.OPTION_SYMBOL_INFO +  nowDate, 10, TimeUnit.DAYS);
        }catch (Exception e){
            LOGGER.error("parseOptionSymbol----->",e);
        }

    }

    /**
     * 校验optionSymbol的key是否存在
     * @param date
     * @return
     */
    private String checkOptionKey(String date,int version){
        try {
            String nowKey=Constant.OPTION_SYMBOL_INFO + date;
            Set<Object> hashKey=redisTemplate.opsForHash().keys(nowKey);
            if(hashKey.size()==0 && version>=0){
                String yesterday=DateUtils.convertDateToString(DateUtils.minusDay(DateUtils.convertStringToDate(date),1));
                version--;
                nowKey=this.checkOptionKey(yesterday,version);
            }
            return nowKey;
        }catch (Exception e){
            LOGGER.error("checkOptionKey--------->"+e);
        }
        return null;
    }



    private String [] subStringSinaData(String context){
        String [] mdInfoArray = context.split("=");
        if(mdInfoArray.length<2){
            return null;
        }
        String mdInfoElement=mdInfoArray[mdInfoArray.length-1];
        String mdInfoStr=mdInfoElement.substring(1, mdInfoElement.length() - 2);
        String [] mdInfos=mdInfoStr.split(",");
        return mdInfos;
    }
    /**
     * 处理深度行情
     * @param fromCLanguageMdInfo
     * @throws Exception
     */
    public void parseDataItem(String fromCLanguageMdInfo) throws Exception {
        //合约code|最新价|持仓量|申卖价五|申卖量五|申卖价四|申卖量四|申卖价三|申卖量三|申卖价二|申卖量二|申卖价一|申卖量一|申买价一|申买量一|申买价二|申买量二|申买价三|申买量三|申买价四|申买量四|申买价五|申买量五|行情时间|最高价|最低价|成交量|成交额|昨收盘|今开盘|涨停价|跌停价
        // LOGGER.info("接受到C++的行情数据：" + fromCLanguageMdInfo);
        /**根据管道符分割**/
        String [] mdDetailArray=fromCLanguageMdInfo.split("\\|");
        String instrumentCode=mdDetailArray[1],lastPrice=mdDetailArray[2],openInterest=mdDetailArray[3],
                askPrice5=mdDetailArray[4],askVolume5=mdDetailArray[5],askPrice4=mdDetailArray[6],askVolume4=mdDetailArray[7],
                askPrice3=mdDetailArray[8],askVolume3=mdDetailArray[9],askPrice2=mdDetailArray[10],askVolume2=mdDetailArray[11],
                askPrice1=mdDetailArray[12],askVolume1=mdDetailArray[13],bidPrice1=mdDetailArray[14],bidVolume1=mdDetailArray[15],
                bidPrice2=mdDetailArray[16],bidVolume2=mdDetailArray[17],bidPrice3=mdDetailArray[18],bidVolume3=mdDetailArray[19],
                bidPrice4=mdDetailArray[20],bidVolume4=mdDetailArray[21],bidPrice5=mdDetailArray[22],bidVolume5=mdDetailArray[23],
                updateTime=mdDetailArray[24],highPrice=mdDetailArray[25],lowPrice=mdDetailArray[26],volume=mdDetailArray[27],
                turnover=mdDetailArray[28],preClosePrice=mdDetailArray[29],openPrice=mdDetailArray[30],upperLimitPrice=mdDetailArray[31],
                LowerLimitPrice=mdDetailArray[32];
        String nowDate=DateUtils.convertDateToString(new Date());
        Object optionSymbolObj=this.beforeOptionSymbolObj(nowDate, instrumentCode, 10);//redisTemplate.opsForHash().get(Constant.OPTION_SYMBOL_INFO + nowDate, instrumentCode);
        OptionSymbol optionSymbol=JSONObject.parseObject(optionSymbolObj.toString(), OptionSymbol.class);
        if(optionSymbol==null){
            LOGGER.error(instrumentCode +"---->optionSymbolIsNotFound");
            return;
        }
        String yesterday=DateUtils.convertDateToString(DateUtils.minusDay(DateUtils.convertStringToDate(nowDate), 1));
        Object yesterdayTickObj=this.beforeTradeday(optionSymbol.getTargetCode(), yesterday, optionSymbol.getInstrumentMonth(), instrumentCode, 10);
        if(yesterdayTickObj!=null){
            OptionMarketDataField yesterdayTick=JSONObject.parseObject(yesterdayTickObj.toString(), OptionMarketDataField.class);
            preClosePrice=yesterdayTick.getLastPrice();
        }else{
            /**如果是新合约 则沿用推过来的最新价**/
            LOGGER.warn(instrumentCode +"---->yesterdayTickIsNotFound");
        }
        Integer remainderDays=DateUtils.daysBetween(new Date(),DateUtils.convertStringToDate(optionSymbol.getExpireDay()));
        /**String deliveryYear, String deliveryMonth, String instrumentMonth, String instrumentType, String targetCode,
         * Integer remainderDays, String instrumentCode, String instrumentName, String lastPrice, String openInterest,
         * String execsPrice, String preClosePrice, String openPrice, String upperLimitPrice, String lowerLimitPrice,
         * String askPrice5, String askVolume5, String askPrice4, String askVolume4, String askPrice3, String askVolume3,
         * String askPrice2, String askVolume2, String askPrice1, String askVolume1, String bidPrice1, String bidVolume1,
         * String bidPrice2, String bidVolume2, String bidPrice3, String bidVolume3, String bidPrice4, String bidVolume4,
         * String bidPrice5, String bidVolume5, String updateTime, String highPrice, String lowPrice, String volume, String turnover**/
        OptionMarketDataField marketDataField=new OptionMarketDataField(
                optionSymbol.getDeliveryYear(),optionSymbol.getDeliveryMonth(),optionSymbol.getInstrumentMonth(),optionSymbol.getInstrumentType(),
                optionSymbol.getTargetCode(),remainderDays,instrumentCode,optionSymbol.getInstrumentName(),lastPrice,openInterest,
                optionSymbol.getExecsPrice(),preClosePrice,openPrice,upperLimitPrice,LowerLimitPrice,
                askPrice5,askVolume5,askPrice4,askVolume4,askPrice3,askVolume3,
                askPrice2,askVolume2,askPrice1,askVolume1,bidPrice1,bidVolume1,
                bidPrice2,bidVolume2,bidPrice3,bidVolume3,bidPrice4,bidVolume4,
                bidPrice5,bidVolume5,updateTime,highPrice,lowPrice,volume,turnover,optionSymbol.getExpireDay()
        );
        //主力合约列表
        SocketJsonData<OptionMarketDataField> socketJsonData=new SocketJsonData<OptionMarketDataField>(Constant.TICK,marketDataField);
        //深度行情json
        Object marketDataFieldToJson= JSONObject.toJSON(socketJsonData);
        //二进制推送 推送压缩后的包
        byte [] data= GZIPUtils.compress(marketDataFieldToJson.toString());
        ByteBuf byteBuf = Unpooled.buffer(data.length * 3);
        byteBuf.writeBytes(data);
//        ChannelGroups.broadcast(new BinaryWebSocketFrame(byteBuf));
        ChannelGroups.writeByMonth(optionSymbol.getInstrumentMonth(),new BinaryWebSocketFrame(byteBuf));
        LOGGER.info(socketJsonData.getData().getInstrumentCode() + ":socket推送的格式：" + marketDataFieldToJson);
//			LOGGER.info("打印推出的数据合约：" + socketJsonData.getData().getInstrumentID() + ",时间：" + socketJsonData.getData().getUpdateTime() + ",最新价：" + socketJsonData.getData().getLastPrice());
        /**主力合约json刷新到redis中**/
        sendRedisLastPrice(optionSymbol.getTargetCode(),nowDate,marketDataField);
    }

    public int randomNum(){
        return (int)((Math.random()*9+1)*10);
    }

    private Object beforeOptionSymbolObj(String nowDate,String instrumentCode,int version){
        try {
            String optionSymbolKey=Constant.OPTION_SYMBOL_INFO + nowDate;
            Object optionSymbolObj=redisTemplate.opsForHash().get(optionSymbolKey,instrumentCode);
            if(optionSymbolObj==null && version>=0){
                String yesterday=DateUtils.convertDateToString(DateUtils.minusDay(DateUtils.convertStringToDate(nowDate), 1));
                version--;
                optionSymbolObj=this.beforeOptionSymbolObj(yesterday, instrumentCode, version);
            }
            return optionSymbolObj;
        }catch (Exception e){
            LOGGER.error("beforeOptionSymbolObj--------->"+e);
        }
        return null;
    }

    private Object beforeTradeday(String targetCode,String date,String instrumentMonth,String instrumentCode,int version){
        try {
            String tickKey=Constant.OPTION_TICKL_INFO+targetCode+":"+instrumentMonth+":"+date;
            Object tickObject=redisTemplate.opsForHash().get(tickKey,instrumentCode);
            if(tickObject==null && version>=0){
                String yesterday=DateUtils.convertDateToString(DateUtils.minusDay(DateUtils.convertStringToDate(date), 1));
                version--;
                tickObject=this.beforeTradeday(targetCode,yesterday,instrumentMonth,instrumentCode,version);
            }
            return tickObject;
        }catch (Exception e){
            LOGGER.error("beforeTradeday--------->",e);
        }
        return null;
    }

    /**
     * 采用线程池去刷新redis最新价
     * @param nowDate
     * @param optionMarketDataField
     */
    public void sendRedisLastPrice(String targetCode,String nowDate,OptionMarketDataField optionMarketDataField) {
        pool.execute(new Thread(() -> {
            try {
                LOGGER.info("sendRedisLastPrice:" + optionMarketDataField.getInstrumentCode() + "," + optionMarketDataField.toString());
                /**主力合约json**/
                Object mainContractToJson= JSONObject.toJSON(optionMarketDataField);
                /**推送完成之后更新首页主力合约列表的合约最新值**/
                redisTemplate.opsForHash().put(Constant.OPTION_TICKL_INFO+targetCode+":"+optionMarketDataField.getInstrumentMonth()+":"+nowDate,optionMarketDataField.getInstrumentCode(),mainContractToJson.toString());
                redisTemplate.expire(Constant.OPTION_TICKL_INFO+targetCode+":"+optionMarketDataField.getInstrumentMonth()+":"+nowDate, 10, TimeUnit.DAYS);
            }catch (Exception e){
                e.printStackTrace();
                LOGGER.warn("sendRedisLastPrice error ..........." + e);
            }
        }));
    }
    private int parseSocketStatus(Exception e)
    {
        if(SocketException.class.isInstance(e))
        {
            String msg = e.getLocalizedMessage().trim();
            if("Connection reset".equalsIgnoreCase(msg))
            {
                return SocketStatusListener.STATUS_RESET;
            }
            else if("Socket is closed".equalsIgnoreCase(msg))
            {
                return SocketStatusListener.STATUS_CLOSE;
            }
            else if("Broken pipe".equalsIgnoreCase(msg))
            {
                return SocketStatusListener.STATUS_PIP_BROKEN;
            }

        }
        return SocketStatusListener.STATUS_UNKOWN;
    }

    /**
     * listen:(这里用一句话描述这个方法的作用). <br/>
     * TODO(这里描述这个方法适用条件 – 可选).<br/>
     *
     */
    public void startListener(SocketStatusListener ssl) {
        listening = true;
        this.socketStatusListener = ssl;
        start();
    }
}
